package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

public class Demo04HBaseFilter {
    Connection conn;
    Table stu;


    public void filterStuAndPrint(Filter filter) throws IOException {
        Scan scan = new Scan();
        scan.setFilter(filter);
        ResultScanner scanner = stu.getScanner(scan);
        for (Result rs : scanner) {
            String id = Bytes.toString(rs.getValue("info".getBytes(), "id".getBytes()));
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));

            System.out.println(id + "," + name + "," + age + "," + gender + "," + clazz);
        }
    }

    @Before
    public void init() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        conn = ConnectionFactory.createConnection(conf);

        TableName stu_t_name = TableName.valueOf("stu");
        stu = conn.getTable(stu_t_name);
    }

    @Test
    public void rowFilterCode() throws IOException {

        BinaryComparator binaryComparator = new BinaryComparator("1500100010".getBytes());

        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.LESS, binaryComparator);


        filterStuAndPrint(rowFilter);

    }

    @Test
    // 基于Family过滤器 对列簇进行过滤
    public void familyFilterCode() throws IOException {
        Scan scan = new Scan();

        String cf = "pass_info";
        BinaryComparator binaryComparator = new BinaryComparator(cf.getBytes());
        RegexStringComparator regexStringComparator = new RegexStringComparator(".*info.*");
        SubstringComparator substringComparator = new SubstringComparator("info");

        FamilyFilter familyFilter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, substringComparator);

        scan.setFilter(familyFilter);

        ResultScanner scanner = conn.getTable(TableName.valueOf("ns1:veh_pass")).getScanner(scan);

        for (Result rs : scanner) {
            String clsd = Bytes.toString(rs.getValue(cf.getBytes(), "clsd".getBytes()));
            String dldm = Bytes.toString(rs.getValue(cf.getBytes(), "dldm".getBytes()));
            String cllx = Bytes.toString(rs.getValue("veh_info".getBytes(), "cllx".getBytes()));
            String hphm = Bytes.toString(rs.getValue("veh_info".getBytes(), "hphm".getBytes()));
            String hpzl = Bytes.toString(rs.getValue("veh_info".getBytes(), "hpzl".getBytes()));


            System.out.println(clsd + "," + dldm + "," + cllx + "," + hphm + "," + hpzl);
        }
    }

    @Test
    public void qualifierFilterCode() throws IOException {
        SubstringComparator substringComparator = new SubstringComparator("e");

        QualifierFilter qualifierFilter = new QualifierFilter(CompareFilter.CompareOp.EQUAL, substringComparator);

        filterStuAndPrint(qualifierFilter);

    }

    @Test
    public void valueFilterCode() throws IOException {
        SubstringComparator substringComparator = new SubstringComparator("文");

        ValueFilter valueFilter = new ValueFilter(CompareFilter.CompareOp.EQUAL, substringComparator);

        filterStuAndPrint(valueFilter);


    }

    @Test
    // 过滤age<22的所有的学生
    public void singleColumnValueFilterCode() throws IOException {
        BinaryComparator binaryComparator = new BinaryComparator("22".getBytes());
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter("info".getBytes(), "age".getBytes(), CompareFilter.CompareOp.LESS, binaryComparator);
        SingleColumnValueFilter singleColumnValueFilter2 = new SingleColumnValueFilter("info".getBytes(), "age".getBytes(), CompareFilter.CompareOp.LESS, "23".getBytes());
        filterStuAndPrint(singleColumnValueFilter2);
    }

    @Test
    // 过滤出文科三班的所有学生
    // 同SingleColumnValueFilter区别在于：
    // SingleColumnValueFilter会返回符合条件的整行数据
    // SingleColumnValueExcludeFilter 会返回符合条件的整行数据，但比较的列除外
    public void singleColumnValueExcludeFilterCode() throws IOException {
        SingleColumnValueExcludeFilter singleColumnValueExcludeFilter = new SingleColumnValueExcludeFilter("info".getBytes(), "clazz".getBytes(), CompareFilter.CompareOp.EQUAL, "文科三班".getBytes());
        filterStuAndPrint(singleColumnValueExcludeFilter);
    }

    @Test
    // 过滤出rowkey以15001008开头的所有的学生
    // PrefixFilter等价于RowFilter+BinaryPrefixComparator
    public void prefixFilterCode() throws IOException {
//        PrefixFilter prefixFilter = new PrefixFilter("15001008".getBytes());
//        filterStuAndPrint(prefixFilter);
        BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("15001008".getBytes());
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);
        filterStuAndPrint(rowFilter);
    }

    @Test
    // 每页返回10条数据，返回第9页的数据
    // 虽然名字里带了Page 但作用和scan的LIMIT没有区别，只可以限定返回的数据条数
    // 需要搭配scan的STARTROW一起使用
    public void pageFilterCode() throws IOException {
        int pageNum = 9;
        int pageSize = 10;
        // 获取第9页的第一条数据的rowkey
        int frontPageSize = (pageNum - 1) * pageSize;

        Scan scan = new Scan();
        scan.setLimit(frontPageSize + 1);

        ResultScanner scanner = stu.getScanner(scan);
        String page9FirstRK = null;
        for (Result rs : scanner) {
            page9FirstRK = Bytes.toString(rs.getRow());
        }

        Scan scan2 = new Scan();
        scan2.withStartRow(page9FirstRK.getBytes());

//        PageFilter pageFilter = new PageFilter(pageSize);
//        scan2.setFilter(pageFilter);
        // 等价于
        scan2.setLimit(pageSize);

        ResultScanner results = stu.getScanner(scan2);
        for (Result rs : results) {
            String id = Bytes.toString(rs.getValue("info".getBytes(), "id".getBytes()));
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));

            System.out.println(id + "," + name + "," + age + "," + gender + "," + clazz);
        }


    }

    @Test
    // HBase中没有专门用于做分页的命令或操作
    // 如果一定要在HBase中实现分页的功能，那么可以通过合理的设计RK来实现
    // 可以直接推导出每一页的RK范围，再结合scan的withStart/StopRow来实现分页功能
    public void pageWithRKDesign() throws IOException {
        int pageNum = 6;
        int pageSize = 9;

        int baseRK = 1500100000;

        int firstRKNums = (pageNum - 1) * pageSize + 1;

        String startRK = String.valueOf(baseRK + firstRKNums);

        String stopRK = String.valueOf(baseRK + firstRKNums + pageSize);


        Scan scan = new Scan();
        scan.withStartRow(startRK.getBytes());
        scan.withStopRow(stopRK.getBytes());

        ResultScanner scanner = stu.getScanner(scan);

        for (Result rs : scanner) {
            String id = Bytes.toString(rs.getValue("info".getBytes(), "id".getBytes()));
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));

            System.out.println(id + "," + name + "," + age + "," + gender + "," + clazz);
        }

    }

    @Test
    // 过滤age<22 并且 clazz等于 '文科四班' 的学生
    public void filterListCode() throws IOException {
        SingleColumnValueFilter filter1 = new SingleColumnValueFilter("info".getBytes(), "age".getBytes(), CompareFilter.CompareOp.LESS, "22".getBytes());
        SingleColumnValueFilter filter2 = new SingleColumnValueFilter("info".getBytes(), "clazz".getBytes(), CompareFilter.CompareOp.EQUAL, "文科四班".getBytes());

        FilterList filterList = new FilterList();
        filterList.addFilter(filter1);
        filterList.addFilter(filter2);

        filterStuAndPrint(filterList);


    }

    @Test
    // 过滤 性别等于男生并且age<22 或者 clazz等于 '文科四班'
    public void filterListCode2() throws IOException {
        SingleColumnValueFilter filter1 = new SingleColumnValueFilter("info".getBytes(), "age".getBytes(), CompareFilter.CompareOp.LESS, "22".getBytes());
        SingleColumnValueFilter filter2 = new SingleColumnValueFilter("info".getBytes(), "clazz".getBytes(), CompareFilter.CompareOp.EQUAL, "文科四班".getBytes());
        SingleColumnValueFilter filter3 = new SingleColumnValueFilter("info".getBytes(), "gender".getBytes(), CompareFilter.CompareOp.EQUAL, "男".getBytes());

        FilterList filterList1 = new FilterList(FilterList.Operator.MUST_PASS_ALL);

        filterList1.addFilter(filter1);
        filterList1.addFilter(filter3);

        FilterList filterList2 = new FilterList(FilterList.Operator.MUST_PASS_ONE);
        filterList2.addFilter(filterList1);
        filterList2.addFilter(filter2);

        filterStuAndPrint(filterList2);

    }



}
